package cn.ac.iie.di.ban.ycl.cleaner.handler;

import cn.ac.iie.di.ban.ycl.cleaner.CleanerMain;
import cn.ac.iie.di.ban.ycl.cleaner.common.Constants;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.REMessage;
import org.apache.commons.configuration2.Configuration;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * @author zmc <zhangmingcheng@iie.ac.cn>
 * @version V2.0
 * @date 2016-9-8 13:53:12
 * @desc 数据去重
 */
public class CleanerHandlerForIvr implements Runnable {

    final Logger LOG = LoggerFactory.getLogger(CleanerHandlerForIvr.class);
    DeduplicationFilterForIvr deduplicationFilter;
    ContentFilter contentFilter;
    CtTimeDataCountEveryDay cTTimeDataCountEveryDay;
    boolean isFilterNecessary;
    boolean isDeduplicationNecessary;

    public CleanerHandlerForIvr(Configuration configuration) {
        deduplicationFilter = new DeduplicationFilterForIvr(configuration);
        contentFilter = new ContentFilter(configuration);
        cTTimeDataCountEveryDay = new CtTimeDataCountEveryDay();
        isFilterNecessary = configuration.getBoolean(Constants.FILTER_NECESSARY);
        isDeduplicationNecessary = configuration.getBoolean(Constants.DEDUPLICATION_NECESSARY);
    }

    @Override
    public void run() {
        while (true) {
            try {
                REMessage message = CleanerMain.handlerQueue.poll(3, TimeUnit.SECONDS);
                if (message == null) {
                    LOG.warn("there is not message from the receiver queue!");
                    continue;
                }
                LOG.info("receive messages num: {}", message.data.size());
                handle(message);
            } catch (Exception ex) {
                LOG.error("clean data has error! {}", ExceptionUtils.getFullStackTrace(ex));
            }
        }
    }

    private void handle(REMessage message) throws Exception {
        //根据ct统计数据量
        Map<String, Integer> ncTimeDataCount = cTTimeDataCountEveryDay.getcTTimeDataCount(message);
        CleanerMain.cTReceiveCountQueue.put(ncTimeDataCount);
        LOG.info("write ct receive count value success!");

        //内容过滤
        if (isFilterNecessary) {
            message = contentFilter.filter(message);
            LOG.info("get after-filter data");
            //垃圾数据
            REMessage rubishMessage = contentFilter.getRubbishMessage();
            LOG.info("get rubbsih");

            CleanerMain.rubbishQueue.put(rubishMessage);//如果队列满，则阻塞
            LOG.info("content filter success!");
        }
        CleanerMain.filterQueue.put(message);//如果队列满，则阻塞

        //修改：清洗后根据ct统计数据量，原来是去重后根据ct统计数据量
        ncTimeDataCount = cTTimeDataCountEveryDay.getcTTimeDataCount(message);
        CleanerMain.cTSendCountQueue.put(ncTimeDataCount);
        LOG.info("write ct send count value success!");

        //数据去重
        if (isDeduplicationNecessary) {
            message = deduplicationFilter.filter(message);
            //交互数据统计
            REMessage ivrCountMsg = deduplicationFilter.getIvrCountMsg();
            CleanerMain.ivrMsgCountQueue.put(ivrCountMsg);

            //发送<_id,ct>
            REMessage ctMsg = deduplicationFilter.getCtMsg();
            CleanerMain.ctQueue.put(ctMsg);

            LOG.info("deduplication success!");
        }
        CleanerMain.deduplicationQueue.put(message);

        /*//根据ct统计数据量
        ncTimeDataCount = cTTimeDataCountEveryDay.getcTTimeDataCount(message);
        CleanerMain.cTSendCountQueue.put(ncTimeDataCount);
        LOG.info("write ct send count value success!");*/
    }
}
